package com.zlm.realtime.dim

/**
 * Author: Harbour
 * Date: 2021-04-18 21:59
 * Desc:
 */

import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONObject}
import com.zlm.realtime.bean.OrderDetail
import com.zlm.realtime.utils.{MyKafkaSinkUtils, MyKafkaUtils, MyPhoenixUtils, MyOffsetUtils}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Author: Harbour
 * Date: 2021-04-17 12:52
 * Desc: 从Kafka获取用户数据存入HBase
 */
object OrderDetailApp {
    def main(args: Array[String]): Unit = {
        // step 1. 获取context
        val conf: SparkConf = new SparkConf().setAppName("OrderDetailApp").setMaster("local[*]")
        val ssc = new StreamingContext(conf, Seconds(5))

        // step 2. 获取kafka输入流
        val topic = "ods_order_detail"
        val groupId = "ods_detail_group"

        val offsetMap: Map[TopicPartition, Long] = MyOffsetUtils.getOffsetFromRedis(topic, groupId)
        val kafkaInputDStream: InputDStream[ConsumerRecord[String, String]] =
            MyKafkaUtils.getKafkaStream(topic, ssc, offsetMap, groupId)

        var offsetRanges = Array.empty[OffsetRange]
        
        val orderDetailDStream: DStream[OrderDetail] = kafkaInputDStream.transform( // step 3. 保存偏移量
            (kafkaRDD: RDD[ConsumerRecord[String, String]]) => {
                offsetRanges = kafkaRDD.asInstanceOf[HasOffsetRanges].offsetRanges
                kafkaRDD
            }
        ).map( // step 4. 转换为order detail
            (record: ConsumerRecord[String, String]) => {
                JSON.parseObject(record.value(), classOf[OrderDetail])
            }
        )

        val orderDetailWithSkuDStream: DStream[OrderDetail] = orderDetailDStream.mapPartitions(
            // step 5. 将订单详情与sku表关联
            (orderDetailIter: Iterator[OrderDetail]) => {
                val orderDetailList: List[OrderDetail] = orderDetailIter.toList

                val skuIdList: List[Long] = orderDetailList.map((_: OrderDetail).sku_id)
                val sql = s"select id ,tm_id,spu_id,category3_id,tm_name ,spu_name,category3_name from mall_sku_info where id in ('${skuIdList.mkString("','")}')"
                val skuInfoMap: Map[String, JSONObject] = MyPhoenixUtils.queryAll(sql).map(
                    (skuJsonObj: JSONObject) => {
                        (skuJsonObj.getString("ID"), skuJsonObj)
                    }
                ).toMap

                for (orderDetail <- orderDetailList) {
                    val skuJsonObj: JSONObject = skuInfoMap.getOrElse(orderDetail.sku_id.toString, null)
                    if (skuJsonObj != null) {
                        orderDetail.spu_id = skuJsonObj.getLong("SPU_ID")
                        orderDetail.spu_name = skuJsonObj.getString("SPU_NAME")
                        orderDetail.tm_id = skuJsonObj.getLong("TM_ID")
                        orderDetail.tm_name = skuJsonObj.getString("TM_NAME")
                        orderDetail.category3_id = skuJsonObj.getLong("CATEGORY3_ID")
                        orderDetail.category3_name = skuJsonObj.getString("CATEGORY3_NAME")
                    }
                    println(orderDetail)
                }

                orderDetailList.toIterator
            }
        )

        // step 6. 将合并后的宽表写入kafka
        orderDetailWithSkuDStream.foreachRDD(
            (orderDetailRDD: RDD[OrderDetail]) => {
                orderDetailRDD.foreach(
                    (orderDetail: OrderDetail) => {
                        MyKafkaSinkUtils.send(
                            "dwd_order_detail",
                            JSON.toJSONString(orderDetail, new SerializeConfig(true))
                        )
                    }
                )
                MyOffsetUtils.saveOffsetToRedis(topic, groupId, offsetRanges)
            }
        )

        ssc.start()
        ssc.awaitTermination()
    }
}

